package com.example.realtimeeb.ods

import com.alibaba.fastjson.JSON
import com.example.realtimeeb.model._
import com.example.realtimeeb.utils.HBaseUtils
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.{Connection, Delete, Put, Table}

/**
 * 自定义实现HBase sink
 */
class SinkHBase extends RichSinkFunction[List[TableObject]] {


  var connection: Connection = _

  var tradeOrderTable: Table = _
  var areaTable: Table = _
  var orderProductTable: Table = _
  var paymentTable:Table = _
  var productCategoryTable: Table = _
  var productInfoTable: Table = _
  var shopAdminOrgTable: Table = _
  var shopTable:Table = _

  override def open(parameters: Configuration): Unit = {
    connection = HBaseUtils.getConnection()

    tradeOrderTable = connection.getTable(TableName.valueOf("lagou_trade_orders"))
    areaTable = connection.getTable(TableName.valueOf("lagou_area"))
    orderProductTable = connection.getTable(TableName.valueOf("lagou_order_product"))
    paymentTable = connection.getTable(TableName.valueOf("lagou_payments"))
    productCategoryTable = connection.getTable(TableName.valueOf("lagou_product_category"))
    productInfoTable = connection.getTable(TableName.valueOf("lagou_product_info"))
    shopAdminOrgTable = connection.getTable(TableName.valueOf("lagou_shop_admin_org"))
    shopTable = connection.getTable(TableName.valueOf("lagou_shops"))

  }

  override def close(): Unit = {
    if (tradeOrderTable != null)
      tradeOrderTable.close()

    if (connection != null)
      connection.close()

  }

  override def invoke(value: List[TableObject], context: SinkFunction.Context[_]): Unit = {
    value.foreach(element => {
      println("sinkhabse:" + element.toString)

      val database: String = element.database
      val tableName: String = element.tableName
      val typeInfo: String = element.typeInfo

      // 将所有表的增删改同步到HBase中

      if (database.equalsIgnoreCase("dw_show")) {

        if (tableName.equalsIgnoreCase("lagou_area")) {
          val area: AreaDetail = JSON.parseObject(element.dataInfo, classOf[AreaDetail])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertArea(areaTable, area)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteArea(areaTable, area)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_trade_orders")) {
          val tradeOrderInfo: TradeOrder = JSON.parseObject(element.dataInfo, classOf[TradeOrder])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertTradeOrder(tradeOrderTable, tradeOrderInfo)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteTradeOrder(tradeOrderTable, tradeOrderInfo)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_order_product")) {
          val orderProduct: OrderProduct = JSON.parseObject(element.dataInfo, classOf[OrderProduct])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertOrderProduct(orderProductTable, orderProduct)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteOrderProduct(orderProductTable, orderProduct)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_payments")) {
          val payment: Payment = JSON.parseObject(element.dataInfo, classOf[Payment])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertPayment(paymentTable, payment)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deletePayment(paymentTable, payment)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_product_category")) {
          val productCategory: ProductCategory = JSON.parseObject(element.dataInfo, classOf[ProductCategory])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertProductCategory(productCategoryTable, productCategory)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteProductCategory(productCategoryTable, productCategory)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_product_info")) {
          val productInfo: ProductInfo = JSON.parseObject(element.dataInfo, classOf[ProductInfo])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertProductInfo(productInfoTable, productInfo)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteProductInfo(productInfoTable, productInfo)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_shop_admin_org")) {
          val shopAdminOrg: ShopAdminOrg = JSON.parseObject(element.dataInfo, classOf[ShopAdminOrg])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertShopAdminOrg(shopAdminOrgTable, shopAdminOrg)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteShopAdminOrg(shopAdminOrgTable, shopAdminOrg)
          }
        }


        if (tableName.equalsIgnoreCase("lagou_shops")) {
          val shop: Shop = JSON.parseObject(element.dataInfo, classOf[Shop])
          if (typeInfo.equalsIgnoreCase("insert")
            || typeInfo.equalsIgnoreCase("update")) {
            insertShop(shopTable, shop)
          } else if (typeInfo.equalsIgnoreCase("delete")) {
            deleteShop(shopTable, shop)
          }
        }
      }
    })

  }

  def insertTradeOrder(hbTable: Table, dataInfo: TradeOrder): Unit = {
    val put = new Put(dataInfo.orderId.getBytes)
    put.addColumn("f1".getBytes, "modifiedTime".getBytes, dataInfo.modifiedTime.getBytes())
    put.addColumn("f1".getBytes, "orderNo".getBytes, dataInfo.orderNo.getBytes())
    put.addColumn("f1".getBytes, "isPay".getBytes, dataInfo.isPay.getBytes())
    put.addColumn("f1".getBytes, "orderId".getBytes, dataInfo.orderId.getBytes())
    put.addColumn("f1".getBytes, "tradeSrc".getBytes, dataInfo.tradeSrc.getBytes())
    put.addColumn("f1".getBytes, "payTime".getBytes, dataInfo.payTime.getBytes())
    put.addColumn("f1".getBytes, "productMoney".getBytes, dataInfo.productMoney.getBytes())
    put.addColumn("f1".getBytes, "totalMoney".getBytes, dataInfo.totalMoney.getBytes())
    put.addColumn("f1".getBytes, "dataFlag".getBytes, dataInfo.dataFlag.getBytes())
    put.addColumn("f1".getBytes, "userId".getBytes, dataInfo.userId.getBytes())
    put.addColumn("f1".getBytes, "areaId".getBytes, dataInfo.areaId.getBytes())
    put.addColumn("f1".getBytes, "createTime".getBytes, dataInfo.createTime.getBytes())
    put.addColumn("f1".getBytes, "payMethod".getBytes, dataInfo.payMethod.getBytes())
    put.addColumn("f1".getBytes, "isRefund".getBytes, dataInfo.isRefund.getBytes())
    put.addColumn("f1".getBytes, "tradeType".getBytes, dataInfo.tradeType.getBytes())
    put.addColumn("f1".getBytes, "status".getBytes, dataInfo.status.getBytes())

    hbTable.put(put)
    println("TradeOrder insert success ")
  }

  def insertArea(hbTable: Table, areaDetail: AreaDetail): Unit = {
    println(areaDetail.toString)
    val put = new Put(areaDetail.id.getBytes())
    put.addColumn("f1".getBytes(), "name".getBytes(), areaDetail.name.getBytes())
    put.addColumn("f1".getBytes(), "pid".getBytes(), areaDetail.pid.getBytes())
    put.addColumn("f1".getBytes(), "sname".getBytes(), areaDetail.sname.getBytes())
    put.addColumn("f1".getBytes(), "level".getBytes(), areaDetail.level.getBytes())
    put.addColumn("f1".getBytes(), "citycode".getBytes(), areaDetail.citycode.getBytes())
    put.addColumn("f1".getBytes(), "yzcode".getBytes(), areaDetail.yzcode.getBytes())
    put.addColumn("f1".getBytes(), "mername".getBytes(), areaDetail.mername.getBytes())
    put.addColumn("f1".getBytes(), "lng".getBytes(), areaDetail.Lng.getBytes())
    put.addColumn("f1".getBytes(), "lat".getBytes(), areaDetail.Lat.getBytes())
    put.addColumn("f1".getBytes(), "pinyin".getBytes(), areaDetail.pinyin.getBytes())

    hbTable.put(put)
    println("area insert success")
  }


  def insertOrderProduct(table: Table, orderProduct: OrderProduct): Unit = {
    val put = new Put(orderProduct.id.getBytes())
    put.addColumn("f1".getBytes(), "orderId".getBytes(), orderProduct.orderId.getBytes())
    put.addColumn("f1".getBytes(), "productId".getBytes(), orderProduct.productId.getBytes())
    put.addColumn("f1".getBytes(), "productNum".getBytes(), orderProduct.productNum.getBytes())
    put.addColumn("f1".getBytes(), "money".getBytes(), orderProduct.money.getBytes())
    put.addColumn("f1".getBytes(), "extra".getBytes(), orderProduct.extra.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), orderProduct.createTime.getBytes())

    table.put(put)
    println("OrderProduct insert success")

  }

  def insertPayment(table: Table, payment: Payment): Unit = {
    val put = new Put(payment.id.getBytes())
    put.addColumn("f1".getBytes(), "payMethod".getBytes(), payment.payMethod.getBytes())
    put.addColumn("f1".getBytes(), "payName".getBytes(), payment.payName.getBytes())
    put.addColumn("f1".getBytes(), "description".getBytes(), payment.description.getBytes())
    put.addColumn("f1".getBytes(), "payOrder".getBytes(), payment.payOrder.getBytes())
    put.addColumn("f1".getBytes(), "online".getBytes(), payment.online.getBytes())

    table.put(put)
    println("Payment insert success")

  }

  def insertProductCategory(table: Table, productCategory: ProductCategory): Unit = {
    val put = new Put(productCategory.catId.getBytes())
    put.addColumn("f1".getBytes(), "parentId".getBytes(), productCategory.parentId.getBytes())
    put.addColumn("f1".getBytes(), "catName".getBytes(), productCategory.catName.getBytes())
    put.addColumn("f1".getBytes(), "isShow".getBytes(), productCategory.isShow.getBytes())
    put.addColumn("f1".getBytes(), "sortNum".getBytes(), productCategory.sortNum.getBytes())
    put.addColumn("f1".getBytes(), "isDel".getBytes(), productCategory.isDel.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), productCategory.createTime.getBytes())
    put.addColumn("f1".getBytes(), "level".getBytes(), productCategory.level.getBytes())

    table.put(put)
    println("ProductCategory insert success")

  }


  def insertProductInfo(table: Table, productInfo: ProductInfo): Unit = {
    val put = new Put(productInfo.productId.getBytes())
    put.addColumn("f1".getBytes(), "productName".getBytes(), productInfo.productName.getBytes())
    put.addColumn("f1".getBytes(), "shopId".getBytes(), productInfo.shopId.getBytes())
    put.addColumn("f1".getBytes(), "price".getBytes(), productInfo.price.getBytes())
    put.addColumn("f1".getBytes(), "isSale".getBytes(), productInfo.isSale.getBytes())
    put.addColumn("f1".getBytes(), "status".getBytes(), productInfo.status.getBytes())
    put.addColumn("f1".getBytes(), "categoryId".getBytes(), productInfo.categoryId.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), productInfo.createTime.getBytes())
    put.addColumn("f1".getBytes(), "modifyTime".getBytes(), productInfo.modifyTime.getBytes())

    table.put(put)
    println("ProductInfo insert success")

  }

  def insertShopAdminOrg(table: Table, shopAdminOrg: ShopAdminOrg): Unit = {
    val put = new Put(shopAdminOrg.id.getBytes())
    put.addColumn("f1".getBytes(), "parentId".getBytes(), shopAdminOrg.parentId.getBytes())
    put.addColumn("f1".getBytes(), "orgName".getBytes(), shopAdminOrg.orgName.getBytes())
    put.addColumn("f1".getBytes(), "orgLevel".getBytes(), shopAdminOrg.orgLevel.getBytes())
    put.addColumn("f1".getBytes(), "isDelete".getBytes(), shopAdminOrg.isDelete.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), shopAdminOrg.createTime.getBytes())
    put.addColumn("f1".getBytes(), "updateTime".getBytes(), shopAdminOrg.updateTime.getBytes())
    put.addColumn("f1".getBytes(), "isShow".getBytes(), shopAdminOrg.isShow.getBytes())
    put.addColumn("f1".getBytes(), "orgType".getBytes(), shopAdminOrg.orgType.getBytes())

    table.put(put)
    println("ShopAdminOrg insert success")

  }

  def insertShop(table: Table, shop: Shop): Unit = {
    val put = new Put(shop.shopId.getBytes())
    put.addColumn("f1".getBytes(), "userId".getBytes(), shop.userId.getBytes())
    put.addColumn("f1".getBytes(), "areaId".getBytes(), shop.areaId.getBytes())
    put.addColumn("f1".getBytes(), "shopName".getBytes(), shop.shopName.getBytes())
    put.addColumn("f1".getBytes(), "shopLevel".getBytes(), shop.shopLevel.getBytes())
    put.addColumn("f1".getBytes(), "status".getBytes(), shop.status.getBytes())
    put.addColumn("f1".getBytes(), "createTime".getBytes(), shop.createTime.getBytes())
    put.addColumn("f1".getBytes(), "modifyTime".getBytes(), shop.modifyTime.getBytes())

    table.put(put)
    println("Shop insert success")

  }

  def deleteArea(hbTable: Table, areaDetail: AreaDetail): Unit = {
    val delete = new Delete(areaDetail.id.getBytes)
    hbTable.delete(delete)
    println("Area delete success")
  }

  def deleteTradeOrder(hbTable: Table, tradeOrder: TradeOrder): Unit = {
    val delete = new Delete(tradeOrder.orderId.getBytes)
    hbTable.delete(delete)
    println("TradeOrder delete success")
  }

  def deleteOrderProduct(hbTable: Table, orderProduct: OrderProduct): Unit = {
    val delete = new Delete(orderProduct.id.getBytes)
    hbTable.delete(delete)
    println("OrderProduct delete success")
  }

  def deletePayment(table: Table, payment: Payment): Unit = {
    val del = new Delete(payment.id.getBytes())
    table.delete(del)
    println("Payment delete success")
  }

  def deleteProductCategory(table: Table, productCategory: ProductCategory): Unit = {
    val del = new Delete(productCategory.catId.getBytes())
    table.delete(del)
    println("ProductCategory delete success")
  }

  def deleteProductInfo(table: Table, productInfo: ProductInfo): Unit = {
    val del = new Delete(productInfo.productId.getBytes())
    table.delete(del)
    println("ProductInfo delete success")
  }

  def deleteShopAdminOrg(table: Table, shopAdminOrg: ShopAdminOrg): Unit = {
    val del = new Delete(shopAdminOrg.id.getBytes())
    table.delete(del)
    println("ShopAdminOrg delete success")
  }

  def deleteShop(table: Table, shop: Shop): Unit = {
    val del = new Delete(shop.shopId.getBytes())
    table.delete(del)
    println("Shop delete success")
  }
}
